package org.niit.service

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

class WeiBatchService {
  // 初始化Spark配置
  private val sparkConf = new SparkConf()
    .setMaster("local[*]") // 生产环境删除此行
    .setAppName("WeiBatchService")
  private val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  spark.sparkContext.setLogLevel("ERROR")
  import spark.implicits._

  /**
   * 入口方法：执行所有离线分析任务
   * @param inputPath 原始数据文件路径（如CSV/Parquet）
   */
  def runAllAnalyses(inputPath: String): Unit = {
    // 读取原始数据（假设为CSV格式，含表头）
    val rawData = spark.read
      .option("header", "true")
      .option("inferSchema", "true")
      .csv(inputPath)
      .cache() // 缓存数据减少重复读取

    // 执行6个离线分析任务
    analysis25(rawData)  // 不同年龄区间用户浏览产品详情页的比例
    analysis26(rawData)  // 不同性别用户浏览支付页面的转化率
    analysis27(rawData)  // 不同设备上用户浏览支付确认页面的比例
    analysis33(rawData)  // 不同操作系统用户的页面顺序偏好
    analysis35(rawData)  // 不同年龄区间在不同设备上的支付行为差异
    analysis36(rawData)  // 不同市场级别下新老用户的整体浏览转化率

    rawData.unpersist() // 释放缓存
    spark.stop()
  }

  /**
   * 25. 分析不同年龄区间用户浏览产品详情页的比例
   */
  private def analysis25(rawData: DataFrame): Unit = {
    // 定义年龄区间
    val ageGroup = when(col("age") <= 20, "1-20")
      .when(col("age").between(21, 30), "21-30")
      .when(col("age").between(31, 40), "31-40")
      .when(col("age").between(41, 50), "41-50")
      .otherwise("51+")

    rawData
      .select(
        ageGroup.alias("age_group"),
        col("product_page").cast("int").alias("viewed") // 1=浏览过详情页，0=未浏览
      )
      .groupBy("age_group")
      .agg(
        count("*").alias("total_users"),
        sum("viewed").alias("viewed_users"),
        // 计算浏览比例：浏览过详情页的用户数 / 总用户数
        (sum("viewed") / count("*")).alias("view_ratio")
      )
      .select("age_group", "view_ratio")
      .orderBy("age_group")
      .write
      .mode("overwrite")
      .csv("output/weiBatch1.csv") // 输出到weiBatch1.csv
  }

  /**
   * 26. 研究不同性别用户浏览支付页面的转化率
   * 转化率定义：浏览支付页的用户数 / 浏览详情页的用户数
   */
  private def analysis26(rawData: DataFrame): Unit = {
    rawData
      .filter(col("product_page").cast("int") === 1) // 仅统计浏览过详情页的用户
      .select(
        col("sex"),
        col("payment_page").cast("int").alias("paid")
      )
      .groupBy("sex")
      .agg(
        count("*").alias("detail_view_users"), // 浏览详情页的用户数
        sum("paid").alias("payment_view_users"), // 浏览支付页的用户数
        // 计算转化率
        (sum("paid") / count("*")).alias("conversion_rate")
      )
      .select("sex", "conversion_rate")
      .write
      .mode("overwrite")
      .csv("output/weiBatch2.csv") // 输出到weiBatch2.csv
  }

  /**
   * 27. 比较不同设备上用户浏览支付确认页面的比例
   * 比例定义：浏览支付确认页的用户数 / 浏览支付页的用户数
   */
  private def analysis27(rawData: DataFrame): Unit = {
    rawData
      .filter(col("payment_page").cast("int") === 1) // 仅统计浏览过支付页的用户
      .select(
        col("device"),
        col("payment_confirmation_page").cast("int").alias("confirmed")
      )
      .groupBy("device")
      .agg(
        count("*").alias("payment_view_users"), // 浏览支付页的用户数
        sum("confirmed").alias("confirmed_users"), // 浏览确认页的用户数
        // 计算比例
        (sum("confirmed") / count("*")).alias("confirmation_ratio")
      )
      .select("device", "confirmation_ratio")
      .write
      .mode("overwrite")
      .csv("output/weiBatch3.csv") // 输出到weiBatch3.csv
  }

  /**
   * 33. 比较不同操作系统用户浏览主页、列表页、产品详情页的顺序偏好
   * 顺序用页面浏览次数的先后比例表示
   */
  private def analysis33(rawData: DataFrame): Unit = {
    // 计算各页面的浏览次数（按操作系统分组）
    rawData
      .groupBy("operative_system")
      .agg(
        sum("home_page").cast("int").alias("home_count"),
        sum("listing_page").cast("int").alias("listing_count"),
        sum("product_page").cast("int").alias("product_count")
      )
      .select(
        col("operative_system"),
        // 计算各页面浏览次数占比（反映顺序偏好）
        col("home_count") / (col("home_count") + col("listing_count") + col("product_count")).alias("home_ratio"),
        col("listing_count") / (col("home_count") + col("listing_count") + col("product_count")).alias("listing_ratio"),
        col("product_count") / (col("home_count") + col("listing_count") + col("product_count")).alias("product_ratio")
      )
      .write
      .mode("overwrite")
      .csv("output/weiBatch4.csv") // 输出到weiBatch4.csv
  }

  /**
   * 35. 研究不同年龄区间用户在不同设备上的支付行为差异
   * 用支付转化率表示：浏览支付页的用户数 / 总用户数
   */
  private def analysis35(rawData: DataFrame): Unit = {
    val ageGroup = when(col("age") <= 20, "1-20")
      .when(col("age").between(21, 30), "21-30")
      .when(col("age").between(31, 40), "31-40")
      .when(col("age").between(41, 50), "41-50")
      .otherwise("51+")

    rawData
      .select(
        ageGroup.alias("age_group"),
        col("device"),
        col("payment_page").cast("int").alias("paid")
      )
      .groupBy("age_group", "device")
      .agg(
        count("*").alias("total_users"),
        sum("paid").alias("payment_users"),
        (sum("paid") / count("*")).alias("payment_rate") // 支付转化率
      )
      .select("age_group", "device", "payment_rate")
      .orderBy("age_group", "device")
      .write
      .mode("overwrite")
      .csv("output/weiBatch5.csv") // 输出到weiBatch5.csv
  }

  /**
   * 36. 比较不同市场级别下新老用户的整体浏览转化率
   * 整体转化率：浏览支付页的用户数 / 浏览主页的用户数
   */

  /**
   * 36. 比较不同市场级别下新老用户的整体浏览转化率
   * 整体转化率：浏览支付页的用户数 / 浏览主页的用户数
   */
  private def analysis36(rawData: DataFrame): Unit = {
    rawData
      .filter(col("home_page").cast("int") === 1) // 仅统计浏览过主页的用户
      .select(
        col("market"),
        col("new_user").cast("int").alias("is_new"), // 1=新用户，0=老用户
        col("payment_page").cast("int").alias("paid")
      )
      .groupBy("market", "is_new")
      .agg(
        count("*").alias("home_users"), // 浏览主页的用户数
        sum("paid").alias("payment_users"), // 浏览支付页的用户数
        (sum("paid") / count("*")).alias("overall_conversion") // 整体转化率
      )
      .select(
        col("market"), // 修正：使用col()包裹
        when(col("is_new") === 1, "新用户").otherwise("老用户").alias("user_type"),
        col("overall_conversion") // 修正：使用col()包裹
      )
      .write
      .mode("overwrite")
      .csv("output/weiBatch6.csv") // 输出到weiBatch6.csv
  }
}